package com.zc.study

import org.apache.flink.api.common.accumulators.LongCounter
import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.core.fs.FileSystem.WriteMode
import collection.JavaConverters._

object DataSetCounter {

  def main(args: Array[String]): Unit = {
    val env = ExecutionEnvironment.getExecutionEnvironment
    val dataSet = env.fromCollection(1 to 16)
    val small5CounterName = "small5Counter"
    val big5CounterName = "big5Counter"

    val data = dataSet.map(new RichMapFunction[Int, Int] {
      val small5Counter = new LongCounter()
      val big5Counter = new LongCounter()

      override def open(parameters: Configuration): Unit = {
        getRuntimeContext.addAccumulator(small5CounterName, small5Counter)
        getRuntimeContext.addAccumulator(big5CounterName, big5Counter)
        println("注册计数器")
      }

      override def map(value: Int): Int = {
        if (value <= 5) {
          small5Counter.add(1)
//          println(s"small5Counter:$value-$small5Counter")
        } else {
          big5Counter.add(1)
//          println(s"big5Counter:$value-$big5Counter")
        }
        value
      }
    })




    val filePath = "Z:\\gitee\\study_imooc\\test-data\\flink\\sink-scala-teacher\\"
    data.writeAsText(filePath,
      WriteMode.OVERWRITE).setParallelism(4)
    val executionResult = env.execute("DataSetCounter")

//    val value1 = executionResult.getAccumulatorResult[Long](small5CounterName)
//    val value2 = executionResult.getAccumulatorResult[Long](big5CounterName)
//    println(s"$small5CounterName:$value1")
//    println(s"$big5CounterName:$value2")

    val accumulatorResults_javaMap = executionResult.getAllAccumulatorResults
    val accumulatorResults_scalaMap = accumulatorResults_javaMap.asScala

    val totalCounter = new LongCounter

    for ((name, counter) <- accumulatorResults_scalaMap) {
      println(s"$name,$counter")
      totalCounter.add(counter.asInstanceOf[Long])
    }
    println(totalCounter)

  }
}
